package com.ebupt.migu.fineoperationkconsumerpush.controller;

import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import com.ebupt.migu.common.pojo.Topic;
import org.apache.commons.codec.digest.DigestUtils;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.support.atomic.RedisAtomicLong;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import com.ebupt.migu.common.push.ApiResponse;
import com.ebupt.migu.common.push.PushRequestBody;
import com.ebupt.migu.common.push.PushTopicBean;
import com.ebupt.migu.common.util.HttpUtils;
import com.ebupt.migu.common.util.JsonUtils;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class KPushConsumerController {

    @Value("${push.ip}")
    private String pushIp;

    @Value("${push.app_key}")
    private String pushAppKey;

    @Value("${push.app_secret}")
    private String pushAppSecret;

    @Value("${push.push_url}")
    private String pushUrl;

    @Resource
    private RedisTemplate redisTemplate;

    @KafkaListener(topics = "RFMENT_PUSH_TOPIC")
    public void consumeMessage(PushTopicBean bean, Acknowledgment ack) {
        // 添加日志追踪requestId
        MDC.put(Topic.REQUEST_ID_KEY,bean.getRequestId());
        // 上报偏移量
        ack.acknowledge();
        try {
            String taskId = bean.getTaskId();
            String url = buildUrl();
            Map<String, Object> params = beanToMap(bean.getBody());
            Map<String, String> header = buildHeader(taskId, bean.getBody());
            long hs = System.currentTimeMillis();
            if (redisTemplate.hasKey(taskId)) {
                sendPostRequest(url, params, header);
            } else {
                ApiResponse response = JsonUtils.json2Obj(sendPostRequest(url, params, header), ApiResponse.class);
                if (response != null && null != response.getData() && !response.getData().isEmpty()) {
                    // 记录 Amber 端返回的 cid
                    String cid = response.getData();
                    redisTemplate.opsForValue().set(taskId, cid, 60 * 60 * 24 * 7, TimeUnit.SECONDS);
                    log.info("{}活动开始下发，Amber 端任务关联的 cid 值为：{}", taskId, cid);
                }
            }
            long he = System.currentTimeMillis();
            log.info("本次http请求耗时：{}ms", he - hs);
            // 记录下发进度
            String strategyId = taskId.substring(0, taskId.length() - 8);
            String date = taskId.substring(taskId.length() - 8);
            String incrKey = strategyId + "_" + date + "_incr";
            if (bean.getBody() != null && bean.getBody().getUid() != null) {
                RedisAtomicLong redisAtomicLong =
                    new RedisAtomicLong(incrKey, Objects.requireNonNull(redisTemplate.getConnectionFactory()));
                redisAtomicLong.addAndGet(bean.getBody().getUid().size());
            }
            long re = System.currentTimeMillis();
            log.info("redis统计耗时：{}ms", re - he);
        } catch (Exception e) {
            log.error("消费消息失败，重试中...", e);
            throw e;
        }
        MDC.remove(Topic.REQUEST_ID_KEY);
    }

    private String buildUrl() {
        return pushIp + pushUrl;
    }

    private Map<String, String> buildHeader(String taskId, PushRequestBody body) {
        Map<String, String> map = new HashMap<>();
        String timestamp = getTimestamp();
        StringBuilder sb = new StringBuilder("MIGU").append("\n");
        sb.append(taskId).append("\n");
        sb.append(body.getPushChannel()).append("\n");
        sb.append(body.getContent().getPushContent()).append("\n");
        sb.append(timestamp).append("\n");
        sb.append(pushAppSecret).append("\n");
        String authorization = pushAppKey + ":" + DigestUtils.md5Hex(sb.toString());
        map.put("Content-Type", "application/json; charset=utf-8");
        map.put("Authorization", authorization);
        map.put("timestamp", timestamp);
        return map;
    }

    private String getTimestamp() {
        return String.valueOf(System.currentTimeMillis());
    }

    private String sendPostRequest(String url, Map<String, Object> params, Map<String, String> header) {
        try {
            return HttpUtils.doPost(url, params, header);
        } catch (Exception e) {
            log.error("发送 HTTP 请求失败", e);
            throw new RuntimeException("发送 HTTP 请求失败", e);
        }
    }

    /**
     * JavaBean 转 Map
     *
     * @param object
     * @return
     */
    public static Map<String, Object> beanToMap(Object object) {
        Map<String, Object> result = new HashMap<>();
        // 获得类的的属性名数组
        Field[] fields = object.getClass().getDeclaredFields();
        try {
            for (Field field : fields) {
                field.setAccessible(true);
                String name = new String(field.getName());
                result.put(name, field.get(object));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return result;
    }
}